package com.brianxia.quickredis.connect;

import com.brianxia.quickredis.connect.stream.RedisInputStream;
import com.brianxia.quickredis.connect.stream.RedisOutputStream;
import com.brianxia.quickredis.constant.Protocol;
import com.brianxia.quickredis.exception.QuickRedisException;
import com.brianxia.quickredis.utils.EncodeUtils;
import com.brianxia.quickredis.utils.IOUtils;
import com.sun.xml.internal.ws.Closeable;

import javax.xml.ws.WebServiceException;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.List;

/**
 * @Description:connection to redis
 * @author: xiadong
 * @Date: 17/5/23
 */
public class Connection implements Closeable {

    private String host = Protocol.host;
    private int port = Protocol.port;
    private Socket socket;
    private RedisOutputStream outputStream;
    private RedisInputStream inputStream;

    public Connection(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public Connection() {

    }

    public boolean isConnected() {
        return socket != null && socket.isBound() && !socket.isClosed() && socket.isConnected()
                && !socket.isInputShutdown() && !socket.isOutputShutdown();
    }

    public void connect() {
        if (!isConnected()) {
            socket = new Socket();
            try {
                socket.setReuseAddress(true);
                socket.setKeepAlive(true);

                socket = new Socket();
                // ->@wjw_add
                socket.setReuseAddress(true);
                socket.setKeepAlive(true); // Will monitor the TCP connection is
                // valid
                socket.setTcpNoDelay(true); // Socket buffer Whetherclosed, to
                // ensure timely delivery of data
                socket.setSoLinger(true, 0); // Control calls close () method,
                // the underlying socket is closed
                // immediately
                // <-@wjw_add

                socket.connect(new InetSocketAddress(host, port), 2000);
                socket.setSoTimeout(2000);

                outputStream = new RedisOutputStream(socket.getOutputStream());
                inputStream = new RedisInputStream(socket.getInputStream());
            } catch (IOException e) {
                throw new QuickRedisException("socket error!");
            }
        }
    }


    private void disconnect() {
        if (isConnected()) {
            try {
                outputStream.flush();
            } catch (IOException e) {
                throw new QuickRedisException("os flush error");
            }
            IOUtils.closeQuietly(socket);
        }
    }

    public void close() throws WebServiceException {
        this.disconnect();
    }


    public void sendCommand(final Protocol.Command cmd, final String... args) {
        connect();

        final byte[][] bytes = new byte[args.length][];
        for (int i = 0; i < args.length; i++) {
            bytes[i] = EncodeUtils.encode(args[i]);
        }


        try {
            outputStream.write((byte) '*');
            outputStream.writeIntCrLf(args.length + 1);
            outputStream.write((byte) '$');
            outputStream.writeIntCrLf(cmd.raw.length);
            outputStream.write(cmd.raw);
            outputStream.writeCrLf();

            for (final byte[] bt : bytes) {
                outputStream.write((byte) '$');
                outputStream.writeIntCrLf(bt.length);
                outputStream.write(bt);
                outputStream.writeCrLf();
            }

            outputStream.flush();
        } catch (IOException e) {
            throw new QuickRedisException(e);
        }

    }


    /*
    For Simple Strings the first byte of the reply is "+"
For Errors the first byte of the reply is "-"
For Integers the first byte of the reply is ":"
For Bulk Strings the first byte of the reply is "$"
For Arrays the first byte of the reply is "*"For Simple Strings the first byte of the reply is "+"
For Errors the first byte of the reply is "-"
For Integers the first byte of the reply is ":"
For Bulk Strings the first byte of the reply is "$"
For Arrays the first byte of the reply is "*"
     */

    public Object getRespFromRedis() {
        //获取首个BYTE判断数据类型b
        final byte b = inputStream.readByte();

        if (b == Protocol.PLUS_BYTE) {
            return processStatusCodeReply(inputStream);
        } else if (b == Protocol.DOLLAR_BYTE) {
            return processBulkReply(inputStream);
        } else if (b == Protocol.ASTERISK_BYTE) {
            return processMultiBulkReply(inputStream);
        } else if (b == Protocol.COLON_BYTE) {
            return processInteger(inputStream);
        } else if (b == Protocol.MINUS_BYTE) {
            processError(inputStream);
            return null;
        } else {
            throw new QuickRedisException("Unknown reply: " + (char) b);
        }

    }


    public String getBulkReply() {
        final byte[] result = (byte[])getRespFromRedis();
        if (null != result) {
            return EncodeUtils.encode(result);
        } else {
            return null;
        }
    }

    private byte[] processStatusCodeReply(final RedisInputStream is) {
        return is.readLineBytes();
    }

    private byte[] processBulkReply(final RedisInputStream is) {
        final int len = is.readIntCrLf();
        if (len == -1) {
            return null;
        }

        final byte[] read = new byte[len];
        int offset = 0;
        while (offset < len) {
            final int size = is.read(read, offset, (len - offset));
            if (size == -1) throw new QuickRedisException(
                    "It seems like server has closed the connection.");
            offset += size;
        }

        // read 2 more bytes for the command delimiter
        is.readByte();
        is.readByte();

        return read;
    }

    private Long processInteger(final RedisInputStream is) {
        return is.readLongCrLf();
    }

    private List<Object> processMultiBulkReply(final RedisInputStream is) {
        final int num = is.readIntCrLf();
        if (num == -1) {
            return null;
        }
        final List<Object> ret = new ArrayList<Object>(num);
        for (int i = 0; i < num; i++) {
            try {
                ret.add(getRespFromRedis());
            } catch (QuickRedisException e) {
                ret.add(e);
            }
        }
        return ret;
    }

    private String processError(final RedisInputStream is) {
        String message = is.readLine();
        return message;
    }
//    public String readReply(){
//        final byte[] resp = (byte[]) readProtocolWithCheckingBroken();
//        if (null == resp) {
//            return null;
//        } else {
//            return SafeEncoder.encode(resp);
//        }
//    }
//
}
